package org.example.config.mqtt;

import com.google.common.base.Splitter;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.example.config.mqtt.property.MqttProperty;
import org.example.constants.SYMBOL;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.Objects;

@Configuration
@EnableIntegration
@ConditionalOnProperty(prefix = "mqtt", name = "enable", havingValue = "true")
@Order(1)
public class MqttConfig {
    @Bean
    public DefaultMqttPahoClientFactory mqttClientFactory(MqttProperty mqttProperty) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(Splitter.on(SYMBOL.COMMA).splitToStream(mqttProperty.getBrokerUrl()).toArray(String[]::new)); // MQTT Broker 地址
        options.setUserName(mqttProperty.getUsername()); // 如果需要认证
        options.setPassword(mqttProperty.getPassword().toCharArray()); // 如果需要认证
        options.setCleanSession(Objects.isNull(mqttProperty.getCleanSession()) ?MqttConnectOptions.CLEAN_SESSION_DEFAULT : mqttProperty.getCleanSession()); // 清理会话
        options.setKeepAliveInterval(Objects.isNull(mqttProperty.getKeepAliveInterval()) ? MqttConnectOptions.KEEP_ALIVE_INTERVAL_DEFAULT : mqttProperty.getKeepAliveInterval()); // 心跳间隔（秒）
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_DEFAULT);
        options.setMaxInflight(Objects.isNull(mqttProperty.getMaxInflight()) ? MqttConnectOptions.MAX_INFLIGHT_DEFAULT : mqttProperty.getMaxInflight());
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(Objects.isNull(mqttProperty.getConnectTimeout()) ? MqttConnectOptions.CONNECTION_TIMEOUT_DEFAULT : mqttProperty.getConnectTimeout());
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MqttMessageConverter messageConverter() {
        JsonPahoMessageConverter jsonPahoMessageConverter = new JsonPahoMessageConverter();
        JsonBytesMessageMapper jsonBytesMessageMapper = new JsonBytesMessageMapper();
        jsonPahoMessageConverter.setBytesMessageMapper(jsonBytesMessageMapper);
        return jsonPahoMessageConverter;
    }

    @Bean("taskScheduler")
    public ThreadPoolTaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(10); // 设置线程池大小
        scheduler.setThreadNamePrefix("mqtt-taskScheduler-"); // 设置线程名前缀
        return scheduler;
    }
}
